//package com.pubinfo.passbook.kafka;
//
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.apache.kafka.clients.producer.*;
//import org.junit.Test;
//
//import java.time.Duration;
//import java.util.Arrays;
//import java.util.Properties;
//
//public class PlainTextClient {
//
//  //主题，请在管理台创建后使用
//  private static final String TOPIC = "test_public_t1";
//  //当前消费实例所属的Consumer Group，请在管理台创建后填写。
//  //属于同一个Consumer Group的消费实例，会负载消费消息。
//  private static final String GROUP = "test_public_g1";
//  //公共接入点地址
//  private static final String BROKERS = "134.108.27.125:8427,134.108.27.124:8427,134.108.27.123:8427";
//
//  @Test
//  public void testProducer() {
//    produce();
//  }
//
//  @Test
//  public void testConsumer() {
//    consume();
//  }
//
//  private void produce() {
//
//    Properties properties = new Properties();
//    properties.put("security.protocol", "PLAINTEXT");
//    properties.put("sasl.mechanism", "PLAIN");
//    properties.put("bootstrap.servers", BROKERS);
//    properties.put("acks", "1");// -1：all ack;1：leader ack;0：none ack
//    properties.put("retries", 3); //失败重试次数
//    properties.put("batch.size", 262144);
//    properties.put("linger.ms", 10);
//    properties.put("buffer.memory", 67108864);
//    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//    //metric.reporters用于上报客户端指标数据，可选（不设置则无法在管理台查询客户端指标信息），建议设置
//    //如未引入ctg-kafka-client-metrics-reporter依赖，以下代码将导致报错
//    properties.put("metric.reporters", "com.ctg.kafka.clients.reporter.KafkaClientMetricsReporter");
//    //每个客户端实例使用不同的client.id(建议使用与应用、业务相关的名称作为前缀)
//    properties.put("client.id", "producer-demo_" + System.currentTimeMillis());
//    //构造Producer对象，注意，该对象是线程安全的，一般情况下，一个进程内可共用一个Producer对象
//    Producer<String, String> producer = new KafkaProducer<String, String>(properties);
//    int i = 0;
//    boolean stopFlag = false;
//    while (!stopFlag) {
//      String message = "hello " + (i++);
//      //如指定key，将会进根key hash到指定分区,没特别需求，建议key为null，由kafka自动分配分区，以免造成各个分区消息分布不均衡
//      String key = null;
//      ProducerRecord record = new ProducerRecord<>(TOPIC, key, message);
//
//      //建议采用callback的方式，获取发送结果
//      producer.send(record, new Callback() {
//        @Override
//        public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
//          if (exception != null) {
//            //todo log and retry
//            System.out.printf("send  exception. %s%n", exception.getMessage());
//            return;
//          }
//          //todo business process...
//          System.out.printf("send message success. topic=%s,key=%s,offset=%s,partition=%d %n",
//              recordMetadata.topic(), key, recordMetadata.offset(), recordMetadata.partition());
//        }
//      });
//    }
//    producer.close();//close
//  }
//
//  private void consume() {
//
//    Properties properties = new Properties();
//    properties.put("security.protocol", "PLAINTEXT");
//    properties.put("sasl.mechanism", "PLAIN");
//    properties.put("bootstrap.servers", BROKERS);
//    properties.put("group.id", GROUP);
//    properties.put("enable.auto.commit", "false");//建议设为false，手动提交
//    properties.put("auto.offset.reset", "earliest");//根据实限情况调整
//    properties.put("max.poll.records", 30);//每次poll的最大记录数，不建议设置过大，以免消费逻辑处理耗时太久
//    //properties.put("max.poll.interval.ms",300000); //两次poll拉取消息调用的最大时时间隔，超过该值，消费实例将会被踢出消费组，当再次调用poll时会重新加入
//    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//    properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//    //如未引入ctg-kafka-client-metrics-reporter依赖，以下代码将导致报错
//    properties.put("metric.reporters", "com.ctg.kafka.clients.reporter.KafkaClientMetricsReporter");
//    //每个客户端实例使用不同的client.id(建议使用与应用、业务相关的名称作为前缀)
//    properties.put("client.id", String.format("%s_%s", GROUP, System.currentTimeMillis()));
//
//    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
//    consumer.subscribe(Arrays.asList(TOPIC));
//    boolean stopFlag = false;
//    while (!stopFlag) {
//      try {
//        //ConsumerRecords<String, String> records = consumer.poll(3000);//deprecated
//        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));
//        System.out.printf("poll records size = %d%n", records.count());
//        for (ConsumerRecord<String, String> record : records) {
//          //todo business process
//          System.out.printf("message:offset = %d, message_key = %s, message_value = %s partition = %s%n", record.offset(), record.key(), record.value(), record.partition());
//        }
//        //consumer.commitSync(); //注意，如果同步commit offset，注意CommitFailedException异常处理
//        consumer.commitAsync();//manual commit
//      } catch (Exception e) {
//        //todo 注意拉取循环内的异常处理，避免跳出拉取循
//      }
//    }
//    consumer.close();//close
//  }
//}
